package fox.framework.rpc.endpoint.exchange;

import fox.framework.rpc.client.message.request.RpcClientRequest;
import fox.framework.rpc.client.message.response.RpcClientResponse;
import fox.framework.rpc.common.StringUtils;
import fox.framework.rpc.exception.RpcRequestTimeOutException;
import fox.framework.rpc.exception.RpcRequestUndoException;
import fox.framework.rpc.message.RpcHeader;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;

/**
 * AQS实现同步与异步线程交互
 *
 * @author cuichao
 * @Description: QueuedSynchronizeExchange
 * @Date: create in 2020/12/17 15:45
 */
public class QueuedSynchronizeExchange implements RpcExchange {

    private final ConcurrentMap<String, Sync> selector;

    public QueuedSynchronizeExchange() {
        selector = new ConcurrentHashMap<>(128);
    }

    @Override
    public Future register(RpcClientRequest request) {
        String requestId = request.getHeader(RpcHeader.REQUEST_ID);
        Sync sync = new Sync(requestId);
        selector.put(requestId, sync);
        return sync;
    }


    public Future register(RpcClientRequest request, long time, TimeUnit unit) throws InterruptedException {
        String requestId = request.getHeader(RpcHeader.REQUEST_ID);
        Sync sync = new Sync(requestId);
        selector.put(requestId, sync);
        return sync;
    }

    @Override
    public void exchange(RpcClientResponse response) {
        String requestId = response.getHeader(RpcHeader.REQUEST_ID);
        if (StringUtils.isEmpty(requestId)) {
            return;
        }
        Sync sync = selector.getOrDefault(requestId, null);
        //request is undo or interrupted by up's
        if (sync == null) {
            return;
        }
        sync.putLock();
        try {
            //recheck sync == null to help GC (不写也没事 狗头狗头)
            sync = selector.getOrDefault(requestId, null);
            if (sync == null) {
                return;
            }
            sync.response = response;
        } finally {
            sync.unlock();
        }
    }


    private class Sync extends AbstractQueuedSynchronizer implements Future {
        //响应结果数据
        private RpcClientResponse response;
        //requestId
        private String requestId;
        //是否执行的取消操作
        private volatile boolean isUndo;
        //请求线程
        private volatile Thread requestThread;
        //取消线程异原因
        private volatile RpcRequestUndoException undoCause;

        public Sync(String requestId) {
            this.requestId = requestId;
            //state 设为-1 防止request线程提获取锁资源
            //只有返回数据线程lock(-1)时才能拿到资源锁
            setState(-1);
        }

        @Override
        public RpcClientResponse get() throws RpcRequestUndoException {
            requestThread = Thread.currentThread();
            this.exchangeLock();
            try {
                selector.remove(requestId);
                if (isUndo && response == null) {
                    throw undoCause;
                }
                return response;
            } finally {
                this.unlock();
            }
        }

        @Override
        public RpcClientResponse get(long time, TimeUnit unit) throws InterruptedException, RpcRequestUndoException, RpcRequestTimeOutException {
            requestThread = Thread.currentThread();
            boolean tryLock = false;
            try {
                tryLock = this.tryExchangeLock(time, unit);
            } catch (InterruptedException e) {
                selector.remove(requestId);
                if (isUndo) {
                    throw undoCause;
                }
                throw e;
            }
            try {
                selector.remove(requestId);
                if (!tryLock) {
                    throw new RpcRequestTimeOutException("time out");
                }
                return response;
            } finally {
                if (this.isLocked()) {
                    this.unlock();
                }
            }
        }

        @Override
        public void undo(Throwable undoCause) {
            this.putLock();
            try {
                //if selector is not contain requestId
                // the response is return
                if (requestThread != null && selector.containsKey(requestId) && !isUndo) {
                    isUndo = true;
                    this.undoCause = new RpcRequestUndoException(undoCause);
                    selector.remove(requestId);
                    requestThread.interrupt();
                }
            } finally {
                this.unlock();
            }
        }


        @Override
        protected boolean tryAcquire(int arg) {
            //表示是返回数据线程获取该资源
            if (arg == -1) {
                if (compareAndSetState(-1, 1)) {
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
            }
            //request线程获取锁
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        @Override
        protected boolean tryRelease(int arg) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        public void putLock() {
            acquire(-1);
        }

        public void exchangeLock() {
            acquire(1);
        }

        public boolean tryExchangeLock(long time, TimeUnit unit) throws InterruptedException {
            return tryAcquireNanos(1, unit.toNanos(time));
        }

        public void unlock() {
            release(1);
        }

        public boolean isLocked() {
            return isHeldExclusively();
        }

        @Override
        protected final boolean isHeldExclusively() {
            return getExclusiveOwnerThread() == Thread.currentThread();
        }

    }
}
